package org.budo.canal.instance.spring;

import java.net.InetSocketAddress;
import java.util.List;

import org.budo.canal.logposition.meta.BudoMetaLogPositionManager;
import org.budo.canal.message.handler.CanalMessageHandler;
import org.budo.canal.message.handler.DefaultCanalMessageHandler;
import org.budo.canal.parse.inbound.mysql.MysqlEventParserFactoryBean;
import org.budo.support.javax.sql.util.JdbcUtil;
import org.budo.support.lang.util.ReflectUtil;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;

import com.alibaba.otter.canal.common.alarm.CanalAlarmHandler;
import com.alibaba.otter.canal.common.alarm.LogAlarmHandler;
import com.alibaba.otter.canal.instance.core.CanalInstance;
import com.alibaba.otter.canal.meta.CanalMetaManager;
import com.alibaba.otter.canal.meta.MemoryMetaManager;
import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.store.CanalEventStore;
import com.alibaba.otter.canal.store.memory.MemoryEventStoreWithBuffer;
import com.alibaba.otter.canal.store.model.BatchMode;
import com.alibaba.otter.canal.store.model.Event;

import lombok.Getter;
import lombok.Setter;

/**
 * @see com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring
 */
@Getter
@Setter
public abstract class AbstractCanalInstanceFactoryBean implements FactoryBean<CanalInstance>, BeanNameAware {
    private BatchMode batchMode = BatchMode.MEMSIZE;

    private int bufferSize = 16 * 1024;

    private String destination;

    private String beanName;

    /**
     * TODO 这些属性对象，不可以让多个 CanalInstance 用同一个
     */
    private CanalMetaManager metaManager = new MemoryMetaManager();

    private CanalAlarmHandler alarmHandler = new LogAlarmHandler();

    private CanalEventStore<Event> eventStore;

    private CanalEventParser eventParser = null;

    private CanalEventSink<List<Entry>> eventSink;

    // 连接相关配置
    private AuthenticationInfo masterInfo;

    private String host;

    private Integer port = 3306;

    private String username;

    private String password = "";

    // 拉取相关配置
    private CanalMessageHandler canalMessageHandler = new DefaultCanalMessageHandler();

    /**
     * 单次拉消息条数
     */
    private Integer batchSize = 9;

    /**
     * 单次拉消息超时
     */
    private Long timeout = 1000L;

    /**
     * 如果未拿到消息的休息时间
     */
    private Long interval = 1000L;

    public CanalEventStore<Event> getEventStore() {
        if (this.eventStore == null) {
            this.eventStore = this.defaultMemoryEventStoreWithBuffer();
        }
        return this.eventStore;
    }

    /**
     * 作为方法，执行时间在 destination 有值之后
     */
    public CanalEventParser getEventParser() {
        if (null != this.eventParser) {
            return this.eventParser;
        }

        MysqlEventParserFactoryBean eventParser = new MysqlEventParserFactoryBean();
        eventParser.setDestination(this.destination());

        if (null != this.getMetaManager()) {
            BudoMetaLogPositionManager logPositionManager = new BudoMetaLogPositionManager(this.getMetaManager());
            eventParser.setLogPositionManager(logPositionManager);
        }

        this.setEventParser(eventParser.build());
        return this.eventParser;
    }

    public void setEventParser(CanalEventParser eventParser) {
        if (eventParser instanceof AbstractEventParser //
                && null == ReflectUtil.getFieldValue(AbstractEventParser.class, "destination", eventParser)) {
            // ((AbstractEventParser) eventParser).setDestination(this.destination());
        }

        this.eventParser = eventParser;
    }

    protected String destination() {
        if (null != this.getDestination()) {
            return this.getDestination();
        }

        return this.getBeanName();
    }

    protected AuthenticationInfo masterInfo() {
        if (null != this.getMasterInfo()) {
            this.checkConnectMysql(this.getMasterInfo());
            return this.getMasterInfo();
        }

        if (null != this.getHost() && null != this.getUsername()) {
            InetSocketAddress address = new InetSocketAddress(this.getHost(), this.getPort());
            AuthenticationInfo authenticationInfo = new AuthenticationInfo(address, this.getUsername(), this.getPassword());

            this.checkConnectMysql(authenticationInfo);
            return authenticationInfo;
        }

        throw new IllegalArgumentException("#105 masterInfo and host or username is  null, this=" + this);
    }

    /**
     * 配置检查
     */
    private void checkConnectMysql(AuthenticationInfo authenticationInfo) {
        JdbcUtil.checkConnectMysql(authenticationInfo.getAddress().getHostName(), //
                authenticationInfo.getAddress().getPort(), //
                authenticationInfo.getUsername(), //
                authenticationInfo.getPassword());
    }

    private MemoryEventStoreWithBuffer defaultMemoryEventStoreWithBuffer() {
        MemoryEventStoreWithBuffer memoryEventStoreWithBuffer = new MemoryEventStoreWithBuffer();
        memoryEventStoreWithBuffer.setRaw(false); // 注意，需要 false
        memoryEventStoreWithBuffer.setDdlIsolation(true);

        memoryEventStoreWithBuffer.setBatchMode(this.getBatchMode()); // 限制内存消耗
        memoryEventStoreWithBuffer.setBufferSize(this.getBufferSize());

        return memoryEventStoreWithBuffer;
    }

    @Override
    public Class<?> getObjectType() {
        return CanalInstance.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }
}
